package com.rem.kafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author Rem
 * @date 2023-01-06
 */
@RestController
public class KafkaController {

    @Autowired
    KafkaTemplate kafkaTemplate;

    /**
     * 异步发送
     */
    @GetMapping("/send")
    public void send() {
        User user = new User(16, "jack");
        kafkaTemplate.send("first", user);
    }

    /**
     * 带回调的发送-同步
     */
    @GetMapping("/send2")
    public void send2() {
        User user = new User(17, "jack");
        kafkaTemplate.send("first", user).addCallback(success -> {
            System.out.println(success);
        }, failure -> {
            System.err.println(failure.getMessage());
        });
    }

    /**
     * 带回调的发送-同步3
     */
    @GetMapping("/send3")
    public void send3() {
        User user = new User(18, "jack");
        kafkaTemplate.send("first", user).addCallback(new ListenableFutureCallback<User>() {
            @Override
            public void onFailure(Throwable ex) {
                System.err.println(ex.getMessage());
            }

            @Override
            public void onSuccess(User result) {
                System.out.println(result);
            }
        });
    }
}
